package com.nucarf.bi.digtalize.function;

import com.alibaba.fastjson.JSON;
import com.nucarf.bi.digtalize.bean.*;
import com.nucarf.bi.digtalize.util.FeiShuWarnUtil;
import com.nucarf.bi.digtalize.util.PropertyUtil;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

import java.util.Date;

public class Map2SinkBeanFunction extends RichFlatMapFunction<BinLogRecord, SinkKafkaBean> {

    private String rechargeTopic = PropertyUtil.getPropertyValue("kafka.topic.recharge");
    private String consumeTopic = PropertyUtil.getPropertyValue("kafka.topic.consume");
    private String balanceTopic = PropertyUtil.getPropertyValue("kafka.topic.balance");

    private String defaultCusCode = PropertyUtil.getPropertyValue("crm.default.cus_code");

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //DaoManager.initialConnection();
    }

    @Override
    public void flatMap(BinLogRecord value, Collector<SinkKafkaBean> out) throws Exception {
        try {
            SinkKafkaBean sinkBean = null;
            String table = value.getTable();
            switch (table) {
                case "global_company_order":
                    // 业务数据转推送数据
                    GlobalConsume globalConsume = (GlobalConsume) value.getAfter();
                    if (!(globalConsume.getStatus() == 1 && globalConsume.getChange_type() == 1)) {
                        sinkBean = globalConsume.copyToSinkKafkaBean(consumeTopic, defaultCusCode);
                        out.collect(sinkBean);
                    } else {
                        String msg = "数据错误: status = 1 && change_type = 1, " + JSON.toJSONString(globalConsume);
                        WarnMessage warnMessage = new WarnMessage(PropertyUtil.getPropertyValue("feishu.warn.evn"), msg, new Date().toString());
                        FeiShuWarnUtil.sendFeishuWarning(warnMessage.getMessage());
                    }
                    sinkBean = globalConsume.copyToSinkKafkaBean(consumeTopic, defaultCusCode);
                    if (sinkBean != null) {
                        out.collect(sinkBean);
                    }
                    break;
                case "global_company_recharge":
                    GlobalRecharge recharge = (GlobalRecharge) value.getAfter();
                    sinkBean = recharge.copyToSinkKafkaBean(rechargeTopic, defaultCusCode);
                    out.collect(sinkBean);
                    break;
                case "global_company_balance":
                    GlobalBalance balance = (GlobalBalance) value.getAfter();
                    sinkBean = balance.copyToSinkKafkaBean(balanceTopic, defaultCusCode);
                    out.collect(sinkBean);
                    break;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close() throws Exception {
        // DaoManager.destroyConnection();
        super.close();
    }

}
